package com.intct.dws;

import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.AreaControl;
import com.intct.hbase.bean.MonitorInfo;
import com.intct.common.Constant;
import com.intct.util.KafkaUtil;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashSet;
import java.util.Set;

/**
 * @author
 * @date
 **/
public class DwsVehicleMainApp {

    public static void main(String[] args) throws Exception {

        /*
        多并行度：每个线程都有一个watermark。且每一个线程都是基于自己接收数据的事件时间最大值。
                因此，导致到最后现在还没获取到最小的watermark，所以window无法被触发执行。
                只有所有的线程的最小watermark都满足watermark时间>= window_end_time时，触发历史窗才会执行。
          0001,1538359882000		2018-10-01 10:11:22
          0002,1538359886000		2018-10-01 10:11:26
          0003,1538359892000		2018-10-01 10:11:32
          0004,1538359893000		2018-10-01 10:11:33
          0005,1538359894000		2018-10-01 10:11:34
          0006,1538359896000		2018-10-01 10:11:36
          0007,1538359897000		2018-10-01 10:11:37
         */

        /*
        代码中设置了并行度为1
        env.setParallelism(1);
        如果这里不设置的话，代码在运行的时候会默认读取本机CPU数量设置并行度。
        下面我们来验证一下，把代码中的并行度调整为2:
        env.setParallelism(2);
        1.发现玄机如下:在第二条事件时，其实已经达到窗的触发时机，但是因为并行度为2，只有等到最小
        2.watermark到的时候才会触发窗计算。发现线程44处理的是001和003，线程42处理的是0002，所以只有等到线程42到达后，水印才会起作用执行2018-10-01 10:11:33.000所在的窗。
        0001,1538359890000		2018-10-01 10:11:30
        0002,1538359903000		2018-10-01 10:11:43
        0003,1538359908000		2018-10-01 10:11:48

        https://blog.csdn.net/qq_21383435/article/details/106792841
        https://cloud.tencent.com/developer/article/1960010
         */

        /*
            watermark多并行度下的传递
                1、接收到上游多个，取最小
                2、往下游多个发送，广播
        */

        // 1.基本环境准备
        //1.1 指定流处理环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2002);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //1.2 设置并行度
        env.setParallelism(1);

        // 启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 2.从kafka的主题中读取业务数据
        //2.1 声明消费的主题以及消费者组
        //2.2 创建消费者对象
        String ckAndGroupId = "dws_vehicle_group_id";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(Constant.KAFAK_TOPIC_DWD_VEHICLE, ckAndGroupId);
        //2.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka_Source");

        // 3.对流中的数据类型进行转换   jsonStr->MonitorInfo
        SingleOutputStreamOperator<MonitorInfo> monitorDS = kafkaStrDS.map(m -> JSONObject.parseObject(m, MonitorInfo.class));

        // ---------------------------Begin EventTime--------------------------------
        // 4.定义 watermark(Flink最大允许的延迟时间/乱序时间为多少)
        SingleOutputStreamOperator<MonitorInfo> withWatermarkDS = monitorDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        // forMonotonousTimestamps 升序的Watermark,没有等待时间
                        // forBoundedOutOfOrderness() 乱序的Watermark
                        .<MonitorInfo>forMonotonousTimestamps()
                        // 指定时间戳分配器，从数据中提取
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<MonitorInfo>() {
                                    @Override
                                    public long extractTimestamp(MonitorInfo monitorInfo, long recordTimestamp) {
                                        long timestamp = monitorInfo.getActionTime() * 1000;
                                        return timestamp;
                                    }
                                }
                        )
        );
        // 4.按照区域进行分组
        KeyedStream<MonitorInfo, String> keyedDS = withWatermarkDS.keyBy(monitorInfo -> monitorInfo.getAreaId());

        // 5.按照车辆区域进行集合
        // 5.1 侧输出流
        OutputTag<MonitorInfo> monitorInfoOutputTag = new OutputTag<MonitorInfo>("monitor-late", Types.POJO(MonitorInfo.class));

        // 5.2 开窗集合
        SingleOutputStreamOperator<AreaControl> areaControlStream = keyedDS
                /*
                    window和windowAll的区别
                    window是KeyStream数据流的方法，其并行度是任意的，也就是最大可以和分组key的数量相同
                    windowAll是DataStream数据流的方法，其并行度只能是1，也就是所有的元素都会被聚合到一个算子任务上，性能极差
                 */
                // 使用事件时间语义的窗口-->EventTime窗口处理 1分钟
                // TumblingProcessingTimeWindows.of(Time.minutes(1))
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .allowedLateness(Time.seconds(2)) // 推迟2s关窗
                .sideOutputLateData(monitorInfoOutputTag)  // 关窗后的迟到数据，放入侧输出流
                .apply(new WindowFunction<MonitorInfo, AreaControl, String, TimeWindow>() {
                    @Override
                    public void apply(String areaId, TimeWindow window, Iterable<MonitorInfo> input, Collector<AreaControl> out) throws Exception {
                        String startTime = DateFormatUtils.format(window.getStart(), "yyyy-MM-dd HH:mm:ss");
                        String endTime = DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
                        //input 里面 是车辆经过信号灯被采集的数据，一个车有可能被采集多次，所以要去重
                        // 第一版：使用set 去重
                        Set<String> hashSet = new HashSet<String>();
                        for (MonitorInfo carInfo : input) {
                            hashSet.add(carInfo.getCar());
                        }
                        AreaControl areaControl = new AreaControl(0, areaId, hashSet.size(), startTime, endTime);
                        out.collect(areaControl);

                    }
                });
        // ---------------------------End EventTime--------------------------------

        // ---------------------------Begin ProcessTime--------------------------------
        // ProcessTime 窗口处理
//        SingleOutputStreamOperator<AreaControl> areaControlStream = filterDS.keyBy(k -> k.getAreaId())
//                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
//                .apply(new WindowFunction<MonitorInfo, AreaControl, String, TimeWindow>() {
//                    @Override
//                    public void apply(String key, TimeWindow window, Iterable<MonitorInfo> input, Collector<AreaControl> out) throws Exception {
//
//                        Set<String> set = new HashSet<String>();
//                        for (MonitorInfo monitorInfo : input) {
//                            if (!set.contains(monitorInfo.getCar())) {
//                                set.add(monitorInfo.getCar());
//                            }
//                        }
//                        String start = DateFormatUtils.format(window.getStart(), "yyyy-MM-dd HH:mm:ss");
//                        String end = DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
//                        out.collect(new AreaControl(null, key, set.size(), start, end));
//                    }
//                });
        // ---------------------------End ProcessTime--------------------------------


        areaControlStream.print().setParallelism(1);

        // TODO 6.维表关联

        // TODO 7.写出数据到HBase

        // 8.执行作业
        env.execute();
    }
}
